package com.bw.realtime.app.dwd;



import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

/*
数据流行：
   数据源在哪里
   1.日志服务器
   2.flume采集到kafka
   3.分流 数据量大  业务复杂 新老用户修复
   4.flink加载 page 页面数据
   5.跳出   只有一个页面的会话    超时
   6.我们定义CEP
7.把规则应用到流上
   8.取出符合规则的流和超时流
   9.存入到kafka
   落盘跳出数据


* */
public class DwdTrafficUserJumpDetail1 {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 2. 状态后端设置
        // TODO 3. 从 kafka dwd_traffic_page_log 主题读取日志数据，封装为流
        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_user_jump_detail";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> pageLog = env.addSource(kafkaConsumer);

        // TODO 4. 转换结构
        SingleOutputStreamOperator<JSONObject> mappedStream = pageLog.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("脏数据:" + value);
                }
            }
        });


        // TODO 5. 设置水位线，用于用户跳出统计


        // TODO 6. 按照 mid 分组






        // TODO 7. 定义 CEP 匹配规则

        // TODO 8. 把 Pattern 应用到流上

        //1.取出符合规则的流   2.取出超时流


        // TODO 9. 提取匹配上的事件以及超时事件









        // TODO 11. 合并两个流并将数据写出到 Kafka


    }

}



















